package com.hugh.commom.delayqueue;

import com.alibaba.fastjson.JSON;
import com.hugh.commom.selector.ChannelFutureEntity;
import com.hugh.commom.zk.client.ZkRpcHandle;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.util.concurrent.DelayQueue;
import java.util.concurrent.Executors;

/**
 * @Author: Hugh
 * @Date: 2020/6/25
 */
@SuppressWarnings("unchecked")
@Component
@Slf4j
public class DelayQueueManager implements CommandLineRunner {

    private DelayQueue<DelayTask> delayQueue = new DelayQueue<>();

    @Autowired
    private ZkRpcHandle zkUtil;

    /**
     * 加入到延时队列中
     *
     * @param task
     */
    public void put(DelayTask task) {
        log.info("加入延时任务：{}", task);
        delayQueue.put(task);
    }

    /**
     * 取消延时任务
     *
     * @param task
     * @return
     */
    public boolean remove(DelayTask task) {
        log.info("取消延时任务：{}", task);
        return delayQueue.remove(task);
    }


    @Override
    public void run(String... args) throws Exception {
        log.info("初始化延时队列");
        Executors.newSingleThreadExecutor().execute(new Thread(this::excuteThread));
    }

    /**
     * 延时任务执行线程
     */
    private void excuteThread() {
        while (true) {
            try {
                DelayTask take = delayQueue.take();
                log.info("****开始执行延迟任务，更新过期响应时间，请求入参是:{}", JSON.toJSON(take));
                ChannelFutureEntity data = (ChannelFutureEntity) take.getData();
                data.setRt(0L);
                zkUtil.upDateZkNodeRt(data);

            } catch (Exception e) {
                e.printStackTrace();
            }
        }

    }
}
